package org.zjt.rxjava.lession.second;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Single;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.observables.ConnectableObservable;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.TimeUnit;



/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-06-20 下午6:36
 * @see
 */
@Slf4j
public class ObservableTest {

    /**
     * ConnectableObservable  contnect 才会发射数据
     * @param args
     */
    @Test
    public void publishTest() throws InterruptedException {
        ConnectableObservable<String> publish = Observable.fromCallable(() -> "11111").publish();
        publish.subscribe(log::info);
        publish.subscribe(log::info);
        TimeUnit.SECONDS.sleep(2);
        publish.connect();
    }

    @Test
    public void firstTest() throws InterruptedException {

        Single<String> first = Observable.fromCallable(() -> "11111").first("Nil");  // Nil为默认值
        first.subscribe((Consumer) o -> log.info(String.valueOf(o)));

        Observable.fromCallable(() -> String.valueOf(1/0)).firstElement()
                .onErrorReturn(exception -> {
                    log.error("firstElement",exception);
                    return "Error";
                }).subscribe(log::info);
    }


    /**
     *   用于资源无限的一直暂停资源流转的方式   每次重新开始
     *
     *
     *   您可以调用其dispose（）方法来停止排放并处理该Observer所用的所有资源。它也有一个isDisposed（）方法，表明它是否已经被处置
     *
     * @throws InterruptedException
     */
    @Test
    public void disposeRefreshTest() throws InterruptedException {
        Observable<Long> seconds = Observable.interval(1, TimeUnit.SECONDS);
        Disposable disposable = seconds.subscribe(l -> System.out.println("Received: " + l));

        //sleep 5 seconds
        TimeUnit.SECONDS.sleep(5);

        //dispose and stop emissions  停止流转
        disposable.dispose();
        log.info("是否已经处理完成，并释放资源:{}",disposable.isDisposed());




        //sleep 5 seconds to prove
        //there are no more emissions
        TimeUnit.SECONDS.sleep(5);
    }


    /**
     * publish 创建一个可连接的 Observable。通过connect()方法开启。
     *
     * 创建一个可连续的发送器
     *
     * @throws InterruptedException
     */
    @Test
    public void disposeContinuousTest() throws InterruptedException {

        ConnectableObservable<Long> publish = Observable.interval(1, TimeUnit.SECONDS).publish();
        publish.connect();

        Disposable subscribe = publish.subscribe(l -> System.out.println("Received 1: " + l));

        TimeUnit.SECONDS.sleep(5);
        subscribe.dispose();


        publish.subscribe(l -> System.out.println("Received 2: " + l));
        TimeUnit.SECONDS.sleep(5);
    }




}
